package hbase.operation;

import hbase.common.MyDelete;
import hbase.common.MyGet;
import hbase.common.MyPut;
import hbase.common.MyScan;
import hbase.tools.Constants;
import hbase.tools.HbaseUtils;
import hbase.tools.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import java.io.IOException;
import java.util.*;

public class DataOperation {

    private static Connection conn = null;
    private static Admin admin = null;

    static{
        try {
            Configuration conf = HBaseConfiguration.create();
            conf.set(Constants.ZK_QUORUM,Constants.ZK_QUORUM_NODES);
            conf.set(Constants.ZK_CLIENT_PORT,Constants.ZK_PORT);
            conn = ConnectionFactory.createConnection(conf);
            admin = conn.getAdmin();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 添加或修改单条数据，使用系统当前时间戳,默认关闭连接
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param value
     */
    public static void addOrModifyData(
            String tableName,
            String rowKey,
            String columnFamily,
            String column,
            String value
    ){
        addOrModifyData(tableName,rowKey,columnFamily,column,value,0,true);
    }

    /**
     * 添加或修改单条数据，使用系统当前时间戳，指定是否关闭连接
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param value
     * @param closeFlag true为关闭连接；false为不关闭连接
     */
    public static void addOrModifyData(
            String tableName,
            String rowKey,
            String columnFamily,
            String column,
            String value,
            boolean closeFlag
    ){
        addOrModifyData(tableName,rowKey,columnFamily,column,value,0,closeFlag);
    }

    /**
     * 添加或修改单条数据，使用指定时间戳，默认关闭连接
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param value
     * @param timestamp
     */
    public static void addOrModifyData(
            String tableName,
            String rowKey,
            String columnFamily,
            String column,
            String value,
            long timestamp
    ){
        addOrModifyData(tableName,rowKey,columnFamily,column,value,timestamp,true);
    }
    /**
     * 添加或修改单条数据，使用指定时间戳，指定是否关闭连接
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param value
     * @param timestamp
     * @param closeFlag true为关闭连接；false为不关闭连接
     */
    public static void addOrModifyData(
            String tableName,
            String rowKey,
            String columnFamily,
            String column,
            String value,
            long timestamp,
            boolean closeFlag
    ){

        if(HbaseUtils.isTableExists(admin,tableName)){
            Table table = null;

            try {
                table = conn.getTable(TableName.valueOf(tableName));
                MyPut put = null;
                if(timestamp==0){
                    put = new MyPut(rowKey);
                    put.addColumn(columnFamily,column,value);
                }else{
                    put = new MyPut(rowKey,timestamp);
                    put.addColumn(columnFamily,column,value,timestamp);
                }

                table.put(put);

            } catch (IOException e) {
                throw new RuntimeException(e);
            } finally {
                if(closeFlag){
                    HbaseUtils.closed(conn,table);
                }
            }
        }else{
            System.out.println(tableName + " 表不存在");
        }
    }

    /**
     * 向不同表中添加多条数据
     * @param datas
     */
    public static void addOrModifyMoreData(List<Map<String,String>> datas){
        for (Map<String, String> data : datas) {
            //初始化一条数据的所有的内容
            String tableName = "";
            String rowKey = "";
            String columnFamily = "";
            String column = "";
            String value = "";
            long timestamp = 0;

            //获取一条数据中的所有信息
            if(data.containsKey(Constants.TABLENAME)){
                tableName = data.get(Constants.TABLENAME);
            }
            if(data.containsKey(Constants.ROWKEY)){
                rowKey = data.get(Constants.ROWKEY);
            }
            if(data.containsKey(Constants.COLUMNFAMILY)){
                columnFamily = data.get(Constants.COLUMNFAMILY);
            }
            if(data.containsKey(Constants.COLUMN)){
                column = data.get(Constants.COLUMN);
            }
            if(data.containsKey(Constants.VALUE)){
                value = data.get(Constants.VALUE);
            }
            if(data.containsKey(Constants.TIMESTAMP)){
                timestamp = Long.parseLong(data.get(Constants.TIMESTAMP));
            }
            //如果信息齐全，可以进行添加
            if(StringUtils.isStringNotEmpty(tableName) && StringUtils.isStringNotEmpty(rowKey) &&
                    StringUtils.isStringNotEmpty(columnFamily) && StringUtils.isStringNotEmpty(column)){
                addOrModifyData(tableName,rowKey,columnFamily,column,value,timestamp,false);
            }else{
                System.out.println("信息不全，无法添加");
            }
        }

        HbaseUtils.closed(conn,admin);
    }

    /**
     * 向指定表中添加多条数据
     * @param tableName
     * @param datas
     */
    public static void addOrModifyMoreData(String tableName, List<Put> datas){
        addOrModifyMoreData(tableName,true,datas);
    }

    /**
     * 向指定表中添加多条数据
     * @param tableName
     * @param datas
     */
    public static void addOrModifyMoreData(String tableName,boolean closeFlag,List<Put> datas){

        if(HbaseUtils.isTableExists(admin,tableName)){
            Table table = null;
            try {
                table = conn.getTable(TableName.valueOf(tableName));
                table.put(datas);
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                if(closeFlag){
                    HbaseUtils.closed(conn,table);
                }
            }
        }else{
            System.out.println(tableName + " 表不存在");
        }
    }

    /**
     * 整表扫描
     * @param tableName
     */
    public static void scanTableData(String tableName){
        scanTableData(tableName,null,null);
    }

    /**
     * 按照列族扫描
     * @param tableName
     * @param columnFamily
     */
    public static void scanTableData(String tableName,String columnFamily){
        scanTableData(tableName,columnFamily,null);
    }

    /**
     * 按列族和列扫描
     * @param tableName
     * @param columnFamily
     * @param column
     */
    public static void scanTableData(String tableName,String columnFamily,String column){

        if(HbaseUtils.isTableExists(admin,tableName)){
            Table table = null;
            try {
                table = conn.getTable(TableName.valueOf(tableName));

                MyScan scan = new MyScan();

                //如果只有列族，就查询此列族下的所有信息
                if(StringUtils.isStringNotEmpty(columnFamily) && StringUtils.isStringEmpty(column)){
                    scan.addFamily(columnFamily);
                }else if (StringUtils.isStringNotEmpty(columnFamily) && StringUtils.isStringNotEmpty(column)){
                    //如果列族和列都存在，就查询这个列下的信息
                    scan.addColumn(columnFamily,column);
                }

                //获取需要查询的所有数据的结果对象
                ResultScanner rs = table.getScanner(scan);
                //把结果对象转成迭代器类型，方便后面使用
                Iterator<Result> results = rs.iterator();
                //遍历所有的结果内容
                while(results.hasNext()){
                    Result result = results.next();
                    HbaseUtils.showCellInfo(result);
                }
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                HbaseUtils.closed(conn,table);
            }
        }else{
            System.out.println(tableName + " 表不存在");
        }
    }

    /**
     * 获取某个表中某个行键的所有信息
     * @param tableName
     * @param rowKey
     */
    public static void getTableData(String tableName,String rowKey){
        getTableData(tableName,rowKey,null,null,0);
    }

    /**
     * 获取某个表中某个行键中某个列族的所有信息
     * @param tableName
     * @param rowKey
     * @param columnFamily
     */
    public static void getTableData(String tableName,String rowKey,String columnFamily){
        getTableData(tableName,rowKey,columnFamily,null,0);
    }

    /**
     * 获取某个表中某个行键中某个列族的多版本的所有信息
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param versions
     */
    public static void getTableData(String tableName,String rowKey,String columnFamily,int versions){
        getTableData(tableName,rowKey,columnFamily,null,versions);
    }

    /**
     * 获取某个表中某个行键中某个列族的某个列的所有信息
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     */
    public static void getTableData(String tableName,String rowKey,String columnFamily,String column){
        getTableData(tableName,rowKey,columnFamily,column,0);
    }

    /**
     * 获取某个表中某个行键中某个列族的某个列的多版本的所有信息
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param versions
     */
    public static void getTableData(String tableName,String rowKey,String columnFamily,String column,int versions){

        if(HbaseUtils.isTableExists(admin,tableName)){
            Table table = null;
            try {
                table = conn.getTable(TableName.valueOf(tableName));

                MyGet get = new MyGet(rowKey);

                //如果只有列族，就查询此列族下的所有信息
                if(StringUtils.isStringNotEmpty(columnFamily) && StringUtils.isStringEmpty(column)){
                    get.addFamily(columnFamily);
                }else if (StringUtils.isStringNotEmpty(columnFamily) && StringUtils.isStringNotEmpty(column)){
                    //如果列族和列都存在，就查询这个列下的信息
                    get.addColumn(columnFamily,column);
                }

                if(versions>1){
                    get = (MyGet) get.readVersions(versions);
                }

                Result result = table.get(get);
                HbaseUtils.showCellInfo(result);
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                HbaseUtils.closed(conn,table);
            }
        }else{
            System.out.println(tableName + " 表不存在");
        }
    }

    /**
     * 获取多行数据信息
     * @param tableName
     * @param gets
     */
    public static void getTableMoreData(String tableName,List<Get> gets){
        if(HbaseUtils.isTableExists(admin,tableName)){
            Table table = null;
            try {
                Result[] results = table.get(gets);
                for (Result result : results) {
                    HbaseUtils.showCellInfo(result);
                }
            } catch (IOException e) {
                throw new RuntimeException(e);
            } finally {
                HbaseUtils.closed(conn,table);
            }
        }
    }

    /**
     * 删除整表的所有数据
     * @param tableName
     */
    public static void deleteTableData(String tableName){
        deleteTableData(tableName,true);
    }
    /**
     * 删除整表的所有数据
     * @param tableName
     */
    public static void deleteTableData(String tableName,boolean closeFlag){
        deleteTableData(tableName,closeFlag,null,null,null,0);
    }

    /**
     * 删除某个表的某一行的所有数据
     * @param tableName
     * @param rowKey
     */
    public static void deleteTableData(String tableName,String rowKey){
        deleteTableData(tableName,true,rowKey);
    }
    /**
     * 删除某个表的某一行的所有数据
     * @param tableName
     * @param rowKey
     */
    public static void deleteTableData(String tableName,boolean closeFlag,String rowKey){
        deleteTableData(tableName,closeFlag,rowKey,null,null,0);
    }

    /**
     * 删除某个列族下的所有内容
     * 与shell不一样的地方，shell不允许删除整列族，API支持此从操作
     * @param tableName
     * @param rowKey
     * @param columnFamily
     */
    public static void deleteTableData(String tableName,String rowKey,String columnFamily){
        deleteTableData(tableName,true,rowKey,columnFamily);
    }
    /**
     * 删除某个列族下的所有内容
     * 与shell不一样的地方，shell不允许删除整列族，API支持此从操作
     * @param tableName
     * @param rowKey
     * @param columnFamily
     */
    public static void deleteTableData(String tableName,boolean closeFlag,String rowKey,String columnFamily){
        deleteTableData(tableName,closeFlag,rowKey,columnFamily,null,0);
    }

    /**
     * 删除某个单元格内最新时间戳的数据
     * 与shell不一样的地方，shell可以使用 deleteAll 命令删除整列，API中不存支持删除整列
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     */
    public static void deleteTableData(String tableName,String rowKey,String columnFamily,String column){
        deleteTableData(tableName,true,rowKey,columnFamily,column);
    }
    /**
     * 删除某个单元格内最新时间戳的数据
     * 与shell不一样的地方，shell可以使用 deleteAll 命令删除整列，API中不存支持删除整列
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     */
    public static void deleteTableData(String tableName,boolean closeFlag,String rowKey,String columnFamily,String column){
        deleteTableData(tableName,closeFlag,rowKey,columnFamily,column,0);
    }

    /**
     * 删除某个单元格内指定时间戳的数据
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param timestamp
     */
    public static void deleteTableData(String tableName,String rowKey,String columnFamily,String column,long timestamp){
        deleteTableData(tableName,true,rowKey,columnFamily,column,timestamp);
    }

    /**
     * 删除某个单元格内指定时间戳的数据
     * @param tableName
     * @param rowKey
     * @param columnFamily
     * @param column
     * @param timestamp
     */
    public static void deleteTableData(String tableName,boolean closeFlag,String rowKey,String columnFamily,String column,long timestamp){

        if(HbaseUtils.isTableExists(admin,tableName)){
            Table table = null;
            try {
                table = conn.getTable(TableName.valueOf(tableName));

                //如果 行键，列族，列都为空，应该清空表
                if(StringUtils.isStringEmpty(rowKey)){
                    TableOperation.modifyTableStatus(false,tableName);
                    admin.truncateTable(TableName.valueOf(tableName),true);
                    //执行完清空表操作后，直接退出此方法
                    return;
                }

                MyDelete delete = new MyDelete(rowKey);

                //如果只有列族，就查询此列族下的所有信息
                if(StringUtils.isStringNotEmpty(columnFamily) && StringUtils.isStringEmpty(column)){
                    if(timestamp>0){
                        delete.addFamily(columnFamily,timestamp);
                    }else{
                        delete.addFamily(columnFamily);
                    }
                }else if (StringUtils.isStringNotEmpty(columnFamily) && StringUtils.isStringNotEmpty(column)){
                    if(timestamp>0){
                        //如果列族和列都存在，就查询这个列下的信息
                        delete.addColumn(columnFamily,column,timestamp);
                    }else{
                        //如果列族和列都存在，就查询这个列下的信息
                        delete.addColumn(columnFamily,column);
                    }
                }

                table.delete(delete);

            }catch (Exception e){
                e.printStackTrace();
            }finally {
                if(closeFlag){
                    HbaseUtils.closed(conn,table);
                }
            }
        }else{
            System.out.println(tableName + " 表不存在");
        }
    }

    /**
     * 删除多条数据
     * @param tableName
     * @param deletes
     */
    public static void deleteTableMoreData(String tableName,List<Delete> deletes){
        if(HbaseUtils.isTableExists(admin,tableName)){
            Table table = null;
            try {
                table.delete(deletes);
            } catch (IOException e) {
                throw new RuntimeException(e);
            } finally {
                HbaseUtils.closed(conn,table);
            }
        }
    }


    public static void main(String[] args) {

//        DataOperation.addOrModifyData("s1","001","cf1","name","tom",11111111111111L);

//        List<Map<String,String>> dataList = new ArrayList<>();
//
//        Map<String,String> dataMap1 = new HashMap<>();
//        Map<String,String> dataMap2 = new HashMap<>();
//        Map<String,String> dataMap3 = new HashMap<>();
//
//        dataMap1.put(Constants.TABLENAME,"s1");
//        dataMap1.put(Constants.ROWKEY,"002");
//        dataMap1.put(Constants.COLUMNFAMILY,"cf2");
//        dataMap1.put(Constants.COLUMN,"gender");
//        dataMap1.put(Constants.VALUE,"man");
//
//        dataMap2.put(Constants.TABLENAME,"s2");
//        dataMap2.put(Constants.ROWKEY,"05");
//        dataMap2.put(Constants.COLUMNFAMILY,"baseInfo");
//        dataMap2.put(Constants.COLUMN,"address");
//        dataMap2.put(Constants.VALUE,"hrb");
//        dataMap2.put(Constants.TIMESTAMP,"11111111111133");
//
//        dataMap3.put(Constants.TABLENAME,"s1");
//        dataMap3.put(Constants.ROWKEY,"003");
//        dataMap3.put(Constants.COLUMNFAMILY,"age");
//
//        dataList.add(dataMap1);
//        dataList.add(dataMap2);
//        dataList.add(dataMap3);
//
//        DataOperation.addOrModifyMoreData(dataList);

//        List<Put> putList = new ArrayList<>();
//        MyPut put1 = new MyPut("100");
//        put1.addColumn("cf1","name","rose");
//        MyPut put2 = new MyPut("002");
//        put2.addColumn("cf2","gender","girl");
//        MyPut put3 = new MyPut("101");
//        put3.addColumn("cf1","salary","10000");
//
//        putList.add(put1);
//        putList.add(put2);
//        putList.add(put3);
//
//        DataOperation.addOrModifyMoreData("s1",putList);

//        DataOperation.scanTableData("s1","cf1","name");
//        DataOperation.getTableData("s1","001","cf1","name",3);
        DataOperation.deleteTableData("s1");

    }

}
